package com.liu.concurrent;


import org.apache.log4j.Logger;

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.*;

/**
 * @author liubo
 * @date  2019/4/25
 */
public class FlowDataFromSocket implements Runnable {
	private static Logger log = Logger.getLogger(FlowDataFromSocket.class);
	// 核心线程数 = CPU数＊2
	private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
	// 线程池最大线程数
	private static final int MAXIMUM_POOL_SIZE = 64;
	// 非核心线程空余存活时间
	private static final int KEEP_ALIVE_TIME = 60;
	// 非核心线程空余存活时间单位 秒
	private static final TimeUnit KEEP_ALIVE_TIMEUNIT = TimeUnit.SECONDS;
	// 任务队列
	private final SynchronousQueue<Runnable> workQueue = new SynchronousQueue<Runnable>();

	private ServerSocket server = null;

	private Socket client = null;
	// 启动的线程数目
	private int startThreadNum = 0;
	// 线程序号
	private int threadNum = 0;
	// socket服务器监听端口号
	private int port = 9999;
	// key字节数组
	private static int keymaxlength = 10000;
	// key字节数组
	private static int vaulemaxlength = 1000000;
	// 客户端全量数据
	public static ConcurrentHashMap<String, LinkedBlockingQueue<Object>> map = new ConcurrentHashMap<String, LinkedBlockingQueue<Object>>();
	// 打印台数据
	ConcurrentHashMap<String, Integer> consoleMap = null;

	public FlowDataFromSocket(int port) {
		this.port = port;
	}

	public FlowDataFromSocket(Socket client) {
		this.client = client;
	}

	@Override
	public void run() {
		// 报文头12位：第一个4位字节代表总长度（不包含报文头的长度）；第二个4位字节代表key长度；第三个4位字节代表value长度。
		byte[] bytes = new byte[4];
		// key字节数组
		byte[] keyByte = new byte[keymaxlength];
		// value字节数组
		byte[] valueByte = new byte[vaulemaxlength];
		// key字节数组-->一条数据的key长度
		int keyLen = 0;
		// value字节数组-->一条数据的value长度
		int valueLen = 0;
		String mapKey = "";
		String value = "";
		int length = 0;
		Integer console = 0;
		consoleMap = new ConcurrentHashMap<String, Integer>();

		try {
			InputStream is = client.getInputStream();
			BufferedInputStream bis = new BufferedInputStream(is);
			DataInputStream dis = new DataInputStream(bis);
			while (true) {
				if (dis.available() != 0) {
					byteArrayToLength(bytes, 4, dis);
					keyLen = byteArrayToLength(bytes, 4, dis);
					valueLen = byteArrayToLength(bytes, 4, dis);
					for (int i = 0; i < keyLen; i++) {
						keyByte[i] = dis.readByte();
					}
					mapKey = new String(keyByte, 0, keyLen);

					for (int i = 0; i < valueLen; i++) {
						valueByte[i] = dis.readByte();
					}
					value = new String(valueByte, 0, valueLen);
					// 因为map已经初始化定义了需要的key值，而socket客户端传输过来的是全量的数据，所以我们只要取出已经初始化定义的key，并把其他的key值抛弃掉。
					if (map.get(mapKey) != null) {
						map.get(mapKey).put(value);
						// 需要的告警模块key值数据
						if (consoleMap.get(mapKey) == null) {
							console = 0;
						} else {
							console = consoleMap.get(mapKey);
						}
						consoleMap.put(mapKey, ++console);
					}
					length++;
				} else { // 一个数据包中的数据已经全部读完
					if (length != 0) {
						log.info("一个数据包接收完毕。数据总长度：" + length);

						// 打印保存需要的告警模块key值数据的map
						Iterator<Entry<String, Integer>> it = consoleMap.entrySet().iterator();
						while (it.hasNext()) {
							Entry<String, Integer> entry = it.next();
							if (entry.getValue() != 0) {
								log.info(entry.getKey() + " " + entry.getValue());
							}
						}
						consoleMap.clear();

						length = 0;
						Thread.sleep(0);
					} else {
						if (isServerClose(client)) {
							break;
						}
						Thread.sleep(100);
					}
				}
			}
		} catch (Exception e) {
			log.error(Thread.currentThread().getName() + "socket服务器读取数据失败");
			e.printStackTrace();
		}
	}

	/**
	 * 判断是否断开连接，断开返回true,没有返回false
	 * 
	 * @param socket
	 * @return
	 */
	public Boolean isServerClose(Socket socket) {
		try {
			// 发送1个字节的紧急数据，默认情况下，服务器端没有开启紧急数据处理，不影响正常通信
			socket.sendUrgentData(0);
			return false;
		} catch (Exception e) {
			return true;
		}
	}

	/**
	 * 报文头的字节数组转换为key,value长度
	 * 
	 * @param bytes
	 * @param max
	 * @param dis
	 * @return
	 */
	public static int byteArrayToLength(byte[] bytes, int max, DataInputStream dis) {
		for (int i = 0; i < max; i++) {
			try {
				bytes[i] = dis.readByte();
			} catch (IOException e) {
				log.error(Thread.currentThread().getName() + "dis.readByte读取数据失败");
				e.printStackTrace();
			}
		}
		return bytes[3] & 0xFF | (bytes[2] & 0xFF) << 8 | (bytes[1] & 0xFF) << 16 | (bytes[0] & 0xFF) << 24;
	}

	public void start() {
		log.info("ServerSocket started...");
		try {
			server = new ServerSocket(port);
			log.info("ServerSocket the listener port is bound : " + port);
			/**
			 * 等待客户端发起请求。 ThreadPoolExecutor 1.发起的请求小于核心线程数，有新任务进来就新建一个线程
			 * 2.发起的请求大于核心线程数，空闲的线程去任务队列里取任务执行
			 * 3.如果任务队列满了，并且池中线程数大于或者等于核心线程数，并且小于最大线程数，会再创建新的线程（非核心线程）执行任务
			 * 4.如果任务队列满了，并且池中线程数大于或者等于最大线程数，会执行饱和策略
			 */
			ThreadPoolExecutor service = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME,
					KEEP_ALIVE_TIMEUNIT, workQueue, new ThreadFactory() {
						@Override
						public Thread newThread(Runnable r) {
							return new Thread(r, "FlowDataFromSocket-thread-" + (++threadNum));
						}
					});
			log.info("socketServer线程池核心线程数:" + CORE_POOL_SIZE + " 最大支持线程数:" + MAXIMUM_POOL_SIZE + " 非核心线程空闲时间:"
					+ MAXIMUM_POOL_SIZE + " 单位:" + KEEP_ALIVE_TIMEUNIT);
			while (true) {
				client = server.accept();
				service.execute(new FlowDataFromSocket(client));
				log.info("The client establishes a connection to the server ,Starting a thread " + (++startThreadNum)
						+ client.getInetAddress() + ":" + port);
			}
		} catch (IOException e) {
			log.error("Server listening fails");
			e.printStackTrace();
		} finally {
			try {
				server.close();// 关闭服务器
				log.info("close ServerSocket");
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public long size(String key) {
		return map.get(key).size();
	}

	public String get(String key) {
		return (String) map.get(key).poll();
	}

	public static void main(String[] args) {
		FlowDataFromSocket ss = new FlowDataFromSocket(9999);
		FlowDataFromSocket.map.put("tcp:flow", new LinkedBlockingQueue<Object>(500000));
		FlowDataFromSocket.map.put("P:flow:ipv4", new LinkedBlockingQueue<Object>(500000));
		FlowDataFromSocket.map.put("P:stat:application", new LinkedBlockingQueue<Object>(500000));
		FlowDataFromSocket.map.put("P:stat:network", new LinkedBlockingQueue<Object>(500000));
		FlowDataFromSocket.map.put("P:stat:link", new LinkedBlockingQueue<Object>(500000));
		FlowDataFromSocket.map.put("P:stat:transport", new LinkedBlockingQueue<Object>(500000));
		ss.start();
	}
}
